package com.bw.ads;

import com.bw.bean.Mn5TM4Wide2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.util.Collections;

public class AsyncIOFunc4_2 extends RichAsyncFunction<Mn5TM4Wide2, Mn5TM4Wide2> {
    Connection conn = null;
    PreparedStatement ps1 = null;

    @Override
    public void open(Configuration parameters) throws Exception {
        Class.forName("com.mysql.jdbc.Driver");
        conn = DriverManager.getConnection("jdbc:mysql://hadoop102:3306/tms01?characterEncoding=utf8", "root", "123456");
        ps1 = conn.prepareStatement("select name from base_region_info where id = ?");
    }

    @Override
    public void close() throws Exception {
        ps1.close();
        conn.close();
    }

    @Override
    public void asyncInvoke(Mn5TM4Wide2 mn5TM4Wide, ResultFuture<Mn5TM4Wide2> resultFuture) throws Exception {
        ps1.setLong(1, mn5TM4Wide.getReceiver_city_id());
        ResultSet rs = ps1.executeQuery();
        if (rs.next()) {
            String name = rs.getString("name");
            if ("市辖区".equals(name)) { //如果是直辖市
                ps1.setLong(1, Long.valueOf(mn5TM4Wide.getReceiver_city_id().toString().substring(0, 2) + "0000"));
                ResultSet rs2 = ps1.executeQuery();
                if (rs2.next()) {
                    name = rs2.getString("name");
                }
            }
            mn5TM4Wide.setReceiver_city_name(name);
        }
        resultFuture.complete(Collections.singleton(mn5TM4Wide));
    }
}
